package com.atbeijing.bigdata.spark.core.req

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Spark03_Req_PageFlow {

    def main(args: Array[String]): Unit = {

        val conf = new SparkConf().setMaster("local[*]").setAppName("HotCategoryTop10")
        val sc = new SparkContext(conf)

        val action = sc.textFile("data/user_visit_action.txt")

        val actionDatas = action.map(
            line => {
                val datas = line.split("_")
                UserVisitAction(
                    datas(0),
                    datas(1).toLong,
                    datas(2),
                    datas(3).toLong,
                    datas(4),
                    datas(5),
                    datas(6).toLong,
                    datas(7).toLong,
                    datas(8),
                    datas(9),
                    datas(10),
                    datas(11),
                    datas(12).toLong
                )
            }
        )

        // TODO 统计分母 (页面ID, Count)
        //某个页面访问总次数(pageId,sum)
        val pageCountMap: Map[Long, Int] = actionDatas.map(
            action => {
                (action.page_id, 1)
            }
        ).reduceByKey(_ + _).collect.toMap

        // 统计分子
        // 将数据按照session进行分组
        actionDatas.cache()
        //(sessionId, Iterable[UserVisitAction])
        val sessionData: RDD[(String, Iterable[UserVisitAction])] = actionDatas.groupBy(_.session_id)

        // 将分组后的数据根据页面的访问时间进行排序(升序)
        // (sessionId, List[(页面id-页面id, 1)])
        val mapDatas: RDD[(String, List[(String, Int)])] = sessionData.mapValues(
            iter => {
                val actions: List[UserVisitAction] = iter.toList.sortBy(_.action_time)
                // 【1，2，3，4，5】
                val ids: List[Long] = actions.map(_.page_id)
                // 【(1,2)，(2,3)，(3,4)，(4,5)】
                val flowids: List[(Long, Long)] = ids.zip(ids.tail)
                // 【（1-2，1）， （2-3，1）】
                flowids.map {
                    case (id1, id2) => {
                        (id1 + "-" + id2, 1)
                    }
                }
            }
        )

        // RDD[(String, Int)] => ("a", 1), ("a", 2)
        // RDD[List[(String, Int)]] => List( ("a",1) ), List(("a", 2))
        val flatDatas: RDD[(String, Int)] = mapDatas.map(_._2).flatMap(list=>list)
        //所有用户的数据(页面id-页面id, sum)
        val pageFlowSum: RDD[(String, Int)] = flatDatas.reduceByKey(_ + _)

        // TODO 计算页面单挑转换率
        // 分子 / 分母
        pageFlowSum.foreach {
            //pageflow某一种跳转A->B      sum次数
            case ( pageflow, sum ) => {
                val ids = pageflow.split("-")
                //ids(0)跳转前的页面id
                val cnt = pageCountMap.getOrElse(ids(0).toLong, 1)
                //A->B跳转的次数/A页面访问总次数就是A->B的转换率
                println(s"页面跳转【${pageflow}】的转换率为 = " + (sum.toDouble/cnt))
            }
        }

        sc.stop()

    }
    case class UserVisitAction(
          date: String,//用户点击行为的日期
          user_id: Long,//用户的ID
          session_id: String,//Session的ID
          page_id: Long,//某个页面的ID
          action_time: String,//动作的时间点
          search_keyword: String,//用户搜索的关键词
          click_category_id: Long,//某一个商品品类的ID
          click_product_id: Long,//某一个商品的ID
          order_category_ids: String,//一次订单中所有品类的ID集合
          order_product_ids: String,//一次订单中所有商品的ID集合
          pay_category_ids: String,//一次支付中所有品类的ID集合
          pay_product_ids: String,//一次支付中所有商品的ID集合
          city_id: Long//城市 id
      )
}
